package org.hadoop.hbase;

import lombok.extern.slf4j.Slf4j;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

import java.util.List;

@Slf4j
public class HBaseOperation {
    private Connection con;

    @Before
    public void before() throws Exception {
        Configuration configuration = HBaseConfiguration.create();
        configuration.set("hbase.zookeeper.property.clientPort", "2181");
        configuration.set("hbase.zookeeper.quorum", "server201");
        con = ConnectionFactory.createConnection(configuration);
    }

    @After
    public void after() throws Exception {
        con.close();
    }

    /**
     * 查询所有表
     */
    @Test
    public void listTables() throws Exception {
        Admin admin = con.getAdmin();
        List<TableDescriptor> list = admin.listTableDescriptors();
        for (TableDescriptor t : list) {
            log.info("tableName:" + t.getTableName().getNameAsString());
        }
    }

    /**
     * 查询指定命名空间的表
     */
    @Test
    public void listNamespaceTables() throws Exception {
        Admin admin = con.getAdmin();
        TableName[] tns = admin.listTableNamesByNamespace("ns1");
        for (TableName tn : tns) {
            log.info("name :" + tn.getNameAsString());
        }
        con.close();
    }

    /**
     * 查询表中的数据
     */
    @Test
    public void queryData() throws Exception {
        Table table = con.getTable(TableName.valueOf("ns1:stud"));
        Scan scan1 = new Scan();
        ResultScanner rs = table.getScanner(scan1);
        for (Result r : rs) {
            for (Cell cell : r.listCells()) {
                log.info("RowKey:" + Bytes.toString(CellUtil.cloneRow(cell)) + "\t"
                        + Bytes.toString(CellUtil.cloneFamily(cell)) + "\t"
                        + Bytes.toString(CellUtil.cloneQualifier(cell)) + "\t"
                        + Bytes.toString(CellUtil.cloneValue(cell)));
            }
            log.info("--------------");
        }
        rs.close();
    }

    /**
     * 创建一个表
     */
    @Test
    public void createTable() throws Exception {
        Admin admin = con.getAdmin();
        TableDescriptor td =
                TableDescriptorBuilder
                        .newBuilder(TableName.valueOf("ns1:person"))//表名
                        .setColumnFamily(ColumnFamilyDescriptorBuilder.of("c"))//列族
                        .build();//创建TableDescriptor
        admin.createTable(td);//创建Table
    }

    /**
     * 删除一个表
     */
    @Test
    public void dropTable() throws Exception {
        Admin admin = con.getAdmin();
        admin.disableTable(TableName.valueOf("ns1:person"));
        admin.deleteTable(TableName.valueOf("ns1:person"));
        admin.close();
    }

    /**
     * 向表中插入数据：(如果key值已经存在，则为修改)
     */
    @Test
    public void putData() throws Exception {
        Table table = con.getTable(TableName.valueOf("ns1:stud"));
        Put put = new Put("K001".getBytes());
        put.addColumn("f".getBytes(), "height".getBytes(),Bytes.toBytes(819L));
        table.put(put);
        table.close();
    }

    /**
     * 根据rowke删除记录：
     */
    @Test
    public void deleteByRowKey() throws Exception {
        Table table = con.getTable(TableName.valueOf("ns1:stud"));
        Delete delete = new Delete(Bytes.toBytes("K002"));
        table.delete(delete);
        table.close();
    }

    /**
     * 删除一个单元格，则需要在Delete中输入一个Column即可：
     */
    @Test
    public void deleteColumn() throws Exception {
        String key = "K001";
        Table table = con.getTable(TableName.valueOf("ns1:stud"));
        boolean boo = table.exists(new Get(Bytes.toBytes(key)));
        if (boo) {
            Delete delete = new Delete(Bytes.toBytes(key));
            //设置需要删除的列即可以只删除一个单元格
            delete.addColumn(Bytes.toBytes("f"), Bytes.toBytes("name"));
            table.delete(delete);
        } else {
            log.info("此rowkey不存在:" + key);
        }
        table.close();
    }
}
